package com.sugon.ohdfs.integration.flink.job.stream.sink;

import com.sugon.ohdfs.integration.flink.domain.TestItem;
import org.apache.flink.core.fs.Path;
import org.apache.flink.formats.avro.AvroWriters;
import org.apache.flink.streaming.api.functions.sink.filesystem.StreamingFileSink;
import org.springframework.beans.factory.annotation.Value;

import java.io.Serializable;

public class BulkAvroStreamSinkJob extends AbstractStreamSinkJob{

    @Value("${hdfs.base-path:hdfs://10.11.8.29:9001/flink/sinkTest/stream/sink/bulk/avro}")
    String basePath;

    @Override
    protected Serializable transform(TestItem item) {
        return item;
    }

    @Override
    protected StreamingFileSink buildSink() {
        return StreamingFileSink
                .forBulkFormat(new Path(basePath), AvroWriters.forReflectRecord(TestItem.class))
                .build();
    }
}
